package net.bwie.realtime.vehicle.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.utils.KafkaUtil;
import net.bwie.realtime.vehicle.bean.MonitorInfo;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.nfa.aftermatch.AfterMatchSkipStrategy;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * @author ASUS
 */

/*
{"actionTime": 1756140251000, "monitorId": "M0157", "areaId": "08", "cameraId": "3", "car": "浙J42004", "areaName": "房山区", "latitude": 40.25356558072009, "roadId": "R047", "speedLimit": 60, "speed": 97, "longitude": 116.76609512056085}
{"actionTime": 1756140262000, "monitorId": "M0158", "areaId": "08", "cameraId": "3", "car": "浙J42004", "areaName": "房山区", "latitude": 40.25368215431124, "roadId": "R047", "speedLimit": 60, "speed": 95, "longitude": 116.76619875423156}
{"actionTime": 1756140275000, "monitorId": "M0159", "areaId": "08", "cameraId": "4", "car": "浙J89521", "areaName": "房山区", "latitude": 40.25379872790238, "roadId": "R047", "speedLimit": 60, "speed": 98, "longitude": 116.76630238790227}
{"actionTime": 1756140288000, "monitorId": "M01510", "areaId": "08", "cameraId": "4", "car": "浙J89521", "areaName": "房山区", "latitude": 40.25391530149353, "roadId": "R047", "speedLimit": 60, "speed":991, "longitude": 116.76640602157298}
{"actionTime": 1756140300000, "monitorId": "M01511", "areaId": "08", "cameraId": "3", "car": "浙J13678", "areaName": "房山区", "latitude": 40.25403187508467, "roadId": "R047", "speedLimit": 60, "speed":982, "longitude": 116.76650965524369}
{"actionTime": 1756140312000, "monitorId": "M01512", "areaId": "08", "cameraId": "3", "car": "浙J13678", "areaName": "房山区", "latitude": 40.25414844867582, "roadId": "R047", "speedLimit": 60, "speed":985, "longitude": 116.7666132889144}
{"actionTime": 1756140324000, "monitorId": "M01513", "areaId": "08", "cameraId": "5", "car": "浙J56890", "areaName": "房山区", "latitude": 40.25426502226696, "roadId": "R047", "speedLimit": 60, "speed":9102, "longitude": 116.76671692258511}
{"actionTime": 1756140335000, "monitorId": "M01514", "areaId": "08", "cameraId": "5", "car": "浙J56890", "areaName": "房山区", "latitude": 40.25438159585811, "roadId": "R047", "speedLimit": 60, "speed":998, "longitude": 116.76682055625582}
{"actionTime": 1756140346000, "monitorId": "M01515", "areaId": "08", "cameraId": "3", "car": "浙J24781", "areaName": "房山区", "latitude": 40.25449816944925, "roadId": "R047", "speedLimit": 60, "speed":979, "longitude": 116.76692418992653}
{"actionTime": 1756140355000, "monitorId": "M01516", "areaId": "08", "cameraId": "3", "car": "浙J24781", "areaName": "房山区", "latitude": 40.2546147430404, "roadId": "R047", "speedLimit": 60, "speed": 93, "longitude": 116.76702782359724}
{"actionTime": 1756140362000, "monitorId": "M01517", "areaId": "08", "cameraId": "4", "car": "浙J90325", "areaName": "房山区", "latitude": 40.25473131663154, "roadId": "R047", "speedLimit": 60, "speed":975, "longitude": 116.76713145726795}
{"actionTime": 1756140368000, "monitorId": "M01518", "areaId": "08", "cameraId": "4", "car": "浙J90325", "areaName": "房山区", "latitude": 40.25484789022269, "roadId": "R047", "speedLimit": 60, "speed":978, "longitude": 116.76723509093866}
{"actionTime": 1756140375000, "monitorId": "M01519", "areaId": "08", "cameraId": "5", "car": "浙J67129", "areaName": "房山区", "latitude": 40.25496446381383, "roadId": "R047", "speedLimit": 60, "speed":992, "longitude": 116.76733872460937}
{"actionTime": 1756140382000, "monitorId": "M01520", "areaId": "08", "cameraId": "5", "car": "浙J67129", "areaName": "房山区", "latitude": 40.25508103740498, "roadId": "R047", "speedLimit": 60, "speed":994, "longitude": 116.76744235828008}
{"actionTime": 1756140390000, "monitorId": "M01521", "areaId": "08", "cameraId": "3", "car": "浙J38572", "areaName": "房山区", "latitude": 40.25519761099612, "roadId": "R047", "speedLimit": 60, "speed":986, "longitude": 116.76754599195079}
{"actionTime": 1756140398000, "monitorId": "M01522", "areaId": "08", "cameraId": "3", "car": "浙J38572", "areaName": "房山区", "latitude": 40.25531418458727, "roadId": "R047", "speedLimit": 60, "speed":989, "longitude": 116.7676496256215}
{"actionTime": 1756140405000, "monitorId": "M01523", "areaId": "08", "cameraId": "4", "car": "浙J42004", "areaName": "房山区", "latitude": 40.25543075817841, "roadId": "R047", "speedLimit": 60, "speed":9100, "longitude": 116.76775325929221}
{"actionTime": 1756140471000, "monitorId": "M01524", "areaId": "08", "cameraId": "4", "car": "浙J42004", "areaName": "房山区", "latitude": 40.25554733176956, "roadId": "R047", "speedLimit": 60, "speed":996, "longitude": 116.76785689296292}
**/
public class DangerousDriverController {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 每 1000ms 开始一次 checkpoint
        env.enableCheckpointing(5000);

        // 获取 kafka 数据源
        DataStream<String> ds1 = KafkaUtil.consumerKafka(env, "dwd_car_end");

//        ds1.print();

        // 转换数据结构为实体类
        SingleOutputStreamOperator<MonitorInfo> ds2 = ds1.map(new MapFunction<String, MonitorInfo>() {
            @Override
            public MonitorInfo map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                return new MonitorInfo(
                        jsonObject.getLong("actionTime"),
                        jsonObject.getString("car"),
                        jsonObject.getString("monitorId"),
                        jsonObject.getString("cameraId"),
                        jsonObject.getInteger("speed"),
                        jsonObject.getString("roadId"),
                        jsonObject.getString("areaId"),
                        jsonObject.getDouble("longitude"),
                        jsonObject.getDouble("latitude"),
                       50);
            }
        });


//        monitor.print();

        // 水位线  允许数据迟到五秒
        SingleOutputStreamOperator<MonitorInfo> ds3 = ds2.assignTimestampsAndWatermarks(WatermarkStrategy
                .<MonitorInfo>forBoundedOutOfOrderness(Duration.ofMinutes(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<MonitorInfo>() {
                    @Override
                    public long extractTimestamp(MonitorInfo monitorInfo, long l) {
                        return monitorInfo.getActionTime();
                    }
                }));


//        ds3.print();

        // 定义Pattern
        Pattern<MonitorInfo, MonitorInfo> pattern = Pattern.<MonitorInfo>begin("first", AfterMatchSkipStrategy.skipPastLastEvent())
                .where(new SimpleCondition<MonitorInfo>() {
                    @Override
                    public boolean filter(MonitorInfo monitorInfo) throws Exception {
                        int limitSpeed = monitorInfo.getSpeedLimit();
                        int speed = monitorInfo.getSpeed();
                        return speed > limitSpeed * 1.2;
                    }
                })
                // TODO 匹配3次满足条件的事件
                .times(3)
                // TODO 满足2分钟内条件
                .within(Time.days(1));

        // CEP
        PatternStream<MonitorInfo> patternStream = CEP.pattern(ds3.keyBy(MonitorInfo::getCar), pattern);

        // 输出数据
        SingleOutputStreamOperator<String> ds4 = patternStream.select(new PatternSelectFunction<MonitorInfo, String>() {
            @Override
            public String select(Map<String, List<MonitorInfo>> map) throws Exception {
                return map.get("first").get(0).getCar();
            }
        });

        ds4.print("涉嫌危险驾驶的车牌号是：");

        env.execute("DangerousDriverController");
    }
}
